package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo26HomeWork {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()

    conf.setMaster("local")
    conf.setAppName("hw")

    val sc = new SparkContext(conf)

    val subjectRDD: RDD[String] = sc.textFile("data/subject.txt")
    val scoreRDD: RDD[String] = sc.textFile("data/score.txt")


    //将分数表转换成kv格式，科目编号作为key
    val scoKVRD: RDD[(String, (String, Double))] = scoreRDD.map(line => {
      val split: Array[String] = line.split(",")
      val id: String = split(0)
      val cId: String = split(1)
      val score: Double = split(2).toDouble
      (cId, (id, score))
    })


    //将科目表转换成kv格式，以科目编号作为key
    val subjectKVRD: RDD[(String, (String, Double))] = subjectRDD.map(line => {
      val split: Array[String] = line.split(",")
      val cId: String = split(0)
      val cName: String = split(1)
      val subjectScore: Double = split(2).toDouble
      (cId, (cName, subjectScore))
    })

    //关联科目表和分数表
    val joinRDD: RDD[(String, ((String, Double), (String, Double)))] = scoKVRD.join(subjectKVRD)

    //整理数据
    val scoreAndSubjectRDD: RDD[(String, String, String, Double, Double)] = joinRDD.map {
      case (cId: String, ((id: String, score: Double), (cName: String, subjectScore: Double))) =>
        (id, cId, cName, score, subjectScore)
    }

    //取出及格的分数
    val jigeScoreRDD: RDD[(String, String, String, Double, Double)] = scoreAndSubjectRDD.filter {
      case (id, cId, cName, score, subjectScore) =>
        score >= subjectScore * 0.6
    }

    //统计每个学生及格的科目数
    val jigeStuRDD: RDD[(String, Int)] = jigeScoreRDD
      .map(line => (line._1, 1))
      .reduceByKey((x, y) => x + y)

    //取出都及格的学生
    val douJIgeStuRDD: RDD[(String, Int)] = jigeStuRDD
      .filter {
        case (id: String, num: Int) => num == 6
      }

    //取出科目都及格学生的学号
    val ids: Array[String] = douJIgeStuRDD
      .map(kv => kv._1)
      .collect() //将RDD转换成数组

    //取出学生的分数
    val resultRDD: RDD[(String, String, String, Double, Double)] = jigeScoreRDD.filter {
      case (id, cId, cName, score, subjectScore) =>
        ids.contains(id)
    }

    resultRDD.foreach(println)

  }

}
